package com.zr.domain.bar.service;

import com.zr.domain.bar.entity.MarketDataEvent;
import com.zr.domain.bar.service.fun.RedisSink;
import com.zr.infrastructure.utils.BarAlgoUtil;
import com.zr.disruptor.impl.DisruptorSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class KLineCalculationService {

    @Resource
    private DisruptorSource disruptorSource;

    public void calBarByFrequency(DataStream<MarketDataEvent> marketDataDataStream) throws Exception {
        // 1 分钟 K 线
        DataStream<MarketDataEvent> oneMinuteKLine = BarAlgoUtil.calculateKLine(marketDataDataStream, Time.minutes(1));
        oneMinuteKLine.addSink(new RedisSink());

        // 5 分钟 K 线
        DataStream<MarketDataEvent> fiveMinuteKLine = BarAlgoUtil.calculateKLine(marketDataDataStream, Time.minutes(5));
        fiveMinuteKLine.addSink(new RedisSink());

        // 15 分钟 K 线
        DataStream<MarketDataEvent> fifteenMinuteKLine = BarAlgoUtil.calculateKLine(marketDataDataStream, Time.minutes(15));
        fifteenMinuteKLine.addSink(new RedisSink());

        // 30 分钟 K 线
        DataStream<MarketDataEvent> thirtyMinuteKLine = BarAlgoUtil.calculateKLine(marketDataDataStream, Time.minutes(30));
        thirtyMinuteKLine.addSink(new RedisSink());

        // 1 小时 K 线
        DataStream<MarketDataEvent> oneHourKLine = BarAlgoUtil.calculateKLine(marketDataDataStream, Time.hours(1));
        oneHourKLine.addSink(new RedisSink());


    }

    public void calBar() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<MarketDataEvent> marketDataDataStream = env.addSource(disruptorSource);
        calBarByFrequency(marketDataDataStream);
//        env.execute("K Line Calculation Job");
    }


}
